package com.bw.gmall.realtime.app.ods;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.logging.Logger;
import java.util.logging.Level;

/*
1.获取ods 中日志数topic_log
2.处理过滤topic_log
3.对新老用户进行处理
               is_news:1
                         状态=null  我要把当前人ts转换成年月日  跟新到状态值
                         状态！=null   判断当前数据的ts 和状态日期是否相同  如果不同跟新数据
               is_new:0
                         状态=null    将ts-1day 跟新到状态中
                         如果有值 不需要处理
4. 把主流数据拆分成5个流   分别存入到5个主题中


1.创建流式环境
2.设置并行度
3.运行jar 向topic_log主题 发送数据
4.从 Kafka 读取主流数据
5.数据清洗，转换结构
6.将脏数据写出到 Kafka 指定主题
 */
public class BaseLogAppShopOds {
    private static final Logger logger = Logger.getLogger(BaseLogAppShopOds.class.getName());
    
    public static void main(String[] args) throws Exception {
        System.out.println("===== BaseLogAppShopOds 开始运行 =====");
        
        // 使用当前时间作为时间戳，确保是最近的数据
        long currentTime = System.currentTimeMillis();
        System.out.println("使用当前时间戳: " + currentTime);
        
        // 创建单个时间戳的数组进行测试
        Long[] arrTs = {currentTime};
        
        // 处理每行数据
          /*              访问店铺行为包含观看店铺自播直播间、观看自制全屏页短视频3秒及以上、
                浏览店铺自制图文3秒及以上、浏览全屏微详情、访问宝贝详情页及店铺其他页面*/
        ArrayList<String> list = new ArrayList<>();
        list.add("o"); //其他有行为
        list.add("a");  //观看店铺自播直播间
        list.add("b");//观看自制全屏页短视频3秒及以上
        list.add("c");//浏览店铺自制图文3秒及以上
        list.add("d");//浏览全屏微详情
        list.add("e");//访问宝贝详情页
        list.add("f"); // 添加f类型，确保BaseLogAppShopDwd能获取所有需要的类型

        for (Long arrT : arrTs) {
            // 直接调用，不再有文件读取异常
            mockDatas(list, arrT);
            try {
                Thread.sleep(500); // 减少等待时间，加速数据生成
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        System.out.println("===== BaseLogAppShopOds 运行结束 =====");
    }

    /*
     * 每个店铺商品id是不同的
    // 生成模拟数据用于测试KeywordTopNApp
     * */
    public static void mockDatas(ArrayList<String> list, Long ts) {
        System.out.println("===== 开始生成模拟数据，时间戳: " + ts + " =====");
        ArrayList<JSONObject> jsonObjects = new ArrayList<>();
        Random random = new Random();

        // 地域编码列表
        String[] regions = {"310000", "320000", "330000", "340000", "350000", "410000", "420000", "430000"};
        // 品牌列表
        String[] brands = {"iPhone", "Samsung", "Xiaomi", "OPPO", "vivo", "Huawei", "Apple", "Lenovo"};
        // 渠道列表
        String[] channels = {"Appstore", "HUAWEI", "OPPO", "vivo", "Xiaomi", "Tencent", "Baidu", "Alibaba"};
        // 机型列表
        String[] models = {"iPhone XS", "Galaxy S21", "Mi 11", "Reno6", "X70", "P50", "Mate40", "Nova9"};
        // 操作系统列表
        String[] osList = {"iOS 14.5", "iOS 15.0", "Android 11", "Android 12", "Android 13", "HarmonyOS 2.0"};
        // 版本号列表
        String[] versions = {"v2.1.1", "v2.1.0", "v2.0.9", "v3.0.0", "v3.0.1"};
        // 页面ID列表 - 调整频率以匹配dws层需求
        String[] pageIds = {"home", "category", "good_detail", "search", "cart", "order_confirm", "payment", "user_center"};
        double[] pageIdWeights = {0.1, 0.1, 0.1, 0.5, 0.05, 0.05, 0.05, 0.05}; // 页面访问权重 - 大幅增加search页面权重到0.5
        // 上一页面ID列表
        String[] lastPageIds = {"home", "category", "search", "recommend", "cart", "", null};
        // 来源类型列表
        String[] sourceTypes = {"home", "promotion", "search", "recommend", "category", "activity", "personal"};
        // 商品ID范围
        int productIdMin = 1000;
        int productIdMax = 9999;
        // 动作类型列表 - 增加dws层需要的addcart和favor动作
        String[] actionTypes = {"click", "addcart", "favor", "view", "share"};
        double[] actionTypeWeights = {0.6, 0.2, 0.1, 0.05, 0.05}; // 动作权重
        // 页面区域列表 - 用于点击分布统计
        String[] pageAreas = {"banner", "recommend", "category", "search_result", "goods_list", "default"};

        // 扩展搜索关键词列表，增加更多可能的搜索词，确保分词器能正确处理
        String[] searchKeywords = {
            "手机", "华为手机", "小米手机", "苹果手机", "iPhone", "三星手机",
            "笔记本电脑", "游戏本", "轻薄本", "联想笔记本", "戴尔笔记本",
            "平板电脑", "iPad", "安卓平板", "华为平板", "小米平板",
            "耳机", "蓝牙耳机", "降噪耳机", "游戏耳机", "头戴式耳机",
            "充电器", "快充充电器", "车载充电器", "无线充电器", "多口充电器",
            "智能手表", "Apple Watch", "华为手表", "小米手表", "运动手表",
            "摄像头", "监控摄像头", "行车记录仪", "单反相机", "微单相机",
            "键盘鼠标", "机械键盘", "无线鼠标", "游戏键盘", "静音键盘",
            "游戏手柄", "Xbox手柄", "PS5手柄", "Switch手柄", "摇杆",
            "显示器", "曲面屏", "4K显示器", "电竞显示器", "办公显示器"
        };

        // 生成500条测试数据，确保有足够的数据进行测试
        int totalCount = 500;
        int searchDataCount = 0;
        
        System.out.println("===== 开始生成" + totalCount + "条测试数据（搜索页面权重0.5）=====");
        
        for (int i = 0; i < totalCount; i++) {
            JSONObject jsonObj = new JSONObject();
            JSONObject common = new JSONObject();
            JSONObject page = new JSONObject();

            // 生成common字段
            common.put("ar", regions[random.nextInt(regions.length)]);
            common.put("ba", brands[random.nextInt(brands.length)]);
            common.put("ch", channels[random.nextInt(channels.length)]);
            common.put("is_new", random.nextDouble() < 0.3 ? "1" : "0"); // 30%新用户
            common.put("md", models[random.nextInt(models.length)]);
            common.put("mid", "mid_test_" + (random.nextInt(100) + 1)); // 100个测试用户ID
            common.put("os", osList[random.nextInt(osList.length)]);
            common.put("uid", "user_test_" + (random.nextInt(50) + 1)); // 50个不同的用户ID
            common.put("vc", versions[random.nextInt(versions.length)]);
            common.put("shop_id", 1); // 固定使用店铺ID 1
            common.put("device_type", random.nextInt(2)); // 0:手机, 1:平板

            // 生成page字段 - 使用加权随机选择页面ID
            page.put("during_time", random.nextInt(30000) + 1000); // 1-31秒
            
            // 使用加权随机选择页面ID
            String pageId = weightedRandomSelect(pageIds, pageIdWeights, random);
            page.put("page_id", pageId);
            
            // 确保包含page字段，这是KeywordTopNApp需要的
            boolean isSearchPage = "search".equals(pageId);
            
            if (isSearchPage) {
                // 如果是搜索页面，生成搜索关键词
                // 随机选择1-3个关键词组合
                int keywordCount = random.nextInt(3) + 1;
                StringBuilder keywordsBuilder = new StringBuilder();
                for (int k = 0; k < keywordCount; k++) {
                    if (k > 0) keywordsBuilder.append(",");
                    keywordsBuilder.append(searchKeywords[random.nextInt(searchKeywords.length)]);
                    // 确保每个关键词都不为空且是有效的
                }
                String searchQuery = keywordsBuilder.toString();
                page.put("item", searchQuery);
                page.put("item_type", "keyword");
                
                searchDataCount++;
                
                // 详细记录生成的搜索数据
                    String logMessage = String.format("[SEARCH DATA] 索引=%d, 用户ID=%s, 设备ID=%s, 关键词=%s", 
                            i, common.getString("uid"), common.getString("mid"), searchQuery);
                    System.out.println("=====" + logMessage + "=====");
                    logger.info(logMessage);
                    
                    // 确保ts字段存在且是长整型
                    if (!jsonObj.containsKey("ts")) {
                        jsonObj.put("ts", ts);
                        System.out.println("[搜索数据] 已添加缺失的ts字段: " + ts);
                    }
                    
                    // 确保common和mid字段存在
                    if (!jsonObj.containsKey("common")) {
                        jsonObj.put("common", common);
                        System.out.println("[搜索数据] 已添加缺失的common字段");
                    }
                    
                    // 验证数据完整性
                    System.out.println("[搜索数据验证] 包含common: " + jsonObj.containsKey("common") + ", 包含page: " + jsonObj.containsKey("page") + ", 包含ts: " + jsonObj.containsKey("ts"));
            } else {
                page.put("item", productIdMin + random.nextInt(productIdMax - productIdMin + 1) + "");
                page.put("item_type", "sku_id");
            }
            
            String lastPage = lastPageIds[random.nextInt(lastPageIds.length)];
            if (lastPage != null && !lastPage.isEmpty()) {
                page.put("last_page_id", lastPage);
            }
            page.put("source_type", sourceTypes[random.nextInt(sourceTypes.length)]);

            // 确保包含BaseLogAppShopDwd需要的page_action类型(a,b,c,e,f)
            String[] validActions = {"a", "b", "c", "e", "f"};
            String pageAction;
            if (random.nextDouble() < 0.8) { // 80%的概率使用有效类型
                pageAction = validActions[random.nextInt(validActions.length)];
            } else {
                // 20%的概率使用其他类型
                pageAction = list.get(random.nextInt(list.size()));
            }
            page.put("page_action", pageAction);

            // 组装数据
            jsonObj.put("common", common);
            jsonObj.put("page", page);
            jsonObj.put("ts", ts);

            // 对于搜索页面，确保包含actions字段，且包含addcart动作
            if (isSearchPage && random.nextDouble() < 0.8) { // 80%的搜索页面包含actions
                JSONArray actions = new JSONArray();
                JSONObject action = new JSONObject();
                action.put("action", "addcart"); // 搜索页面添加购物车动作
                action.put("item", productIdMin + random.nextInt(productIdMax - productIdMin + 1) + "");
                action.put("item_type", "sku_id");
                action.put("ts", ts + random.nextInt(30000)); // 相对于日志时间的偏移
                actions.add(action);
                jsonObj.put("actions", actions);
            } else if (random.nextDouble() < 0.6) { // 60%的非搜索页面包含actions
                JSONArray actions = new JSONArray();
                int actionCount = random.nextInt(3) + 1; // 1-3个动作
                for (int j = 0; j < actionCount; j++) {
                    JSONObject action = new JSONObject();
                    // 使用加权随机选择动作类型
                    String actionType = weightedRandomSelect(actionTypes, actionTypeWeights, random);
                    action.put("action", actionType);
                    action.put("item", productIdMin + random.nextInt(productIdMax - productIdMin + 1) + "");
                    action.put("item_type", "sku_id");
                    action.put("ts", ts + random.nextInt(30000)); // 相对于日志时间的偏移
                    
                    // 为click动作添加area字段，支持点击分布统计
                    if ("click".equals(actionType)) {
                        action.put("area", pageAreas[random.nextInt(pageAreas.length)]);
                    }
                    actions.add(action);
                }
                jsonObj.put("actions", actions);
            }

            jsonObjects.add(jsonObj);
        }

        Map<String, Object> config = new HashMap<>();
        config.put("bootstrap.servers", "hadoop102:9092");
        config.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        config.put("acks", "all"); // 确保消息被确认
        config.put("retries", 3); // 设置重试次数

        // 创建Kafka生产者
        Producer<String, String> producer = new KafkaProducer<>(config);

        try {
            int sentSuccessCount = 0;
            int sentFailedCount = 0;
            
            for (JSONObject jsonObject : jsonObjects) {
                try {
                    // 发送到Kafka
                    producer.send(new ProducerRecord<>("ods_traffic", jsonObject.toJSONString()));
                    sentSuccessCount++;
                    
                    // 对于搜索数据，打印更详细的信息
                    JSONObject page = jsonObject.getJSONObject("page");
                    if (page != null && "search".equals(page.getString("page_id"))) {
                        System.out.println("已发送搜索数据到Kafka: 关键词=" + page.getString("item"));
                    }
                } catch (Exception e) {
                    sentFailedCount++;
                    System.err.println("发送数据失败: " + e.getMessage());
                    logger.log(Level.SEVERE, "发送数据失败", e);
                }
            }
            
            System.out.println("数据发送统计: 总数据=" + totalCount + ", 搜索数据=" + searchDataCount + ", 发送成功=" + sentSuccessCount + ", 发送失败=" + sentFailedCount);
            System.out.println("数据发送到Kafka主题: ods_traffic");

        } catch (Exception e) {
            System.err.println("发送数据到Kafka时发生异常: " + e.getMessage());
            e.printStackTrace();
        } finally {
            producer.close();
            System.out.println("Kafka生产者已关闭");
        }

        System.out.println("数据生成完成");
    }
    
    /**
     * 加权随机选择
     * @param options 选项数组
     * @param weights 权重数组，与选项一一对应
     * @param random 随机数生成器
     * @return 选择的选项
     */
    private static String weightedRandomSelect(String[] options, double[] weights, Random random) {
        double totalWeight = 0;
        for (double weight : weights) {
            totalWeight += weight;
        }
        
        double randomValue = random.nextDouble() * totalWeight;
        double cumulativeWeight = 0;
        
        for (int i = 0; i < options.length; i++) {
            cumulativeWeight += weights[i];
            if (randomValue < cumulativeWeight) {
                return options[i];
            }
        }
        
        // 默认返回第一个选项
        return options[0];
    }
}
